package cn.ai.boot.config;

import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

/**
 * @author jie.zhao@dms365.com
 * @Date 2025/6/28 11:50
 */
public class Test333 {

    private static final Logger LOGGER = LoggerFactory.getLogger(Test333.class);

    public static final ExecutorService executorService = Executors.newFixedThreadPool(10);

    public static void main(String[] args) {
        List<Long> updatedCompIds = new ArrayList<>();
        for (int i = 0; i < 500000; i++) {
            updatedCompIds.add(Long.valueOf(i));
        }
        try {

            updateInitializationStatusNew(updatedCompIds);
            System.out.println("============");

        } catch (Exception e) {
            System.out.println("updateInitializationStatusNew.err");
        }

        System.out.println("111");
        executorService.shutdown();
    }

    private static void updateInitializationStatusNew(List<Long> updatedCompIds) {
        List<List<Long>> updatedCompIdsPartition = Lists.partition(updatedCompIds, 10000);
        CountDownLatch countDownLatch = new CountDownLatch(updatedCompIdsPartition.size());
        Semaphore semaphore = new Semaphore(5); // 控制最大并发数量

        for (List<Long> partIds : updatedCompIdsPartition) {
            if (CollectionUtils.isEmpty(partIds)) {
                countDownLatch.countDown();
                continue;
            }
            try {
                semaphore.acquire();
            } catch (Exception e) {
                Thread.currentThread().interrupt();
                countDownLatch.countDown();
//                throw new RuntimeException("Update initialization status failed due to interruption", e);
                continue;
            }

            executorService.execute(() -> {
                try {
                    List<Map<String, Object>> dataList = new ArrayList<>(partIds.size());
                    for (Long id : partIds) {
                        Map<String, Object> component = new HashMap<>(2);
                        component.put("MONGO_ID", id);
                        component.put("PCM_INITIALIZATION_STATUS", "INITIALIZED");
                        dataList.add(component);
                    }
                    batchUpdateCbbData(dataList);
                } catch (Exception e) {
                    LOGGER.error("batchUpdateCbbData.err");
                    throw new RuntimeException("batchUpdateCbbData", e);
                } finally {
                    semaphore.release();
                    countDownLatch.countDown();
                }
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Update initialization status failed due to interruption", e);
        } catch (Exception e) {
            throw new RuntimeException("Exception", e);
        }
    }

    public static void batchUpdateCbbData(List<Map<String, Object>> dataList) {
        System.out.println("batchUpdateCbbData 被调用，数据条数：" + dataList.size() + "mongodb" + dataList.getFirst());
        try {
            Map<String, Object> first = dataList.getFirst();
            if (Integer.valueOf(String.valueOf(first.get("MONGO_ID"))) > 100000) {
                throw new RuntimeException("11123");
            }
            Thread.sleep(2000L);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
